package com.shujia.spark.sql

import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo03WordCnt {
  def main(args: Array[String]): Unit = {

    // 通过SQL以及DSL的方式完成WordCount

    // 构建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .getOrCreate()

    // 构建DF
    val lineDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "|")
      .schema("line String")
      .load("spark/data/words/")

    // 第一种：SQL方式
    lineDF.createOrReplaceTempView("words")

    spark.sql(
      """
        |select  t1.word
        |        ,count(*) as cnt
        |from (
        |   select explode(split(line,",")) as word
        |   from words
        |) t1 group by word
        |""".stripMargin)
    //      .show()


    // 第二种方式：DSL

    /**
     * DSL有两种表达方式：
     * 1、字符串表达式，基于String
     * 2、列表达式，基于Column
     * 优先推荐列表达式
     */

    val stuDF: DataFrame = spark.read.format("csv").option("sep", ",")
      .schema("id String,name String,age Int,gender String,clazz String")
      .load("spark/data/stu/students.txt")

    // 过滤年龄 > 22的学生
    // 字符串表达式
    stuDF.where("age > 22").show()
    // 字符串表达式如果写错了并不会立马报错，需要运行时才能知道，而且所有的操作以及函数都需要开发人员熟记
//    stuDF.where("age dfadsfa> 22").show()

    // 导入SparkSession的隐式转换
    import spark.implicits._ // 回在当前代码中导入隐式转换 $方法，可以将字符串转为Column类型
    // 列表达式：在运行之前就会通过语法检查代码是否正确，能够借助idea代码编辑器进行语法提示
    stuDF.where($"age" > 22).show()


    import org.apache.spark.sql.functions._ // SparkSQL中的函数包，包含了所有Spark内置的函数

    // wordCount基于DSL列表达式
    lineDF
      .select(explode(split($"line",",")) as "words")
      .groupBy($"words")
      .agg(count("*") as "cnt")
      .show()


  }

}
